package org.timewindow

import org.FlinkStreamApp
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector
import org.bean.MinMaxTemp
import org.diysource.SensorSource

/**
 * description ：计算窗口中最大温度和最小温度，附加上窗口结束时间
 * 全窗口聚合函数需要将窗口中的所有元素都缓存下来，如果元素很多，那么存储压力就很大
 * author      ：剧情再美终是戏 
 * mail        : 13286520398@163.com
 * date        ：Created in 2020/2/23 10:32
 * modified By ：
 * version:    : 1.0
 */
object ProcessWindowFunctionExample extends FlinkStreamApp {
  override def doSomeThing(environment: StreamExecutionEnvironment) = {
    // 获取源数据
    val source = environment.addSource(new SensorSource)

    // 计算
    val minTempStream = source
      .map(l => (l.id, l.temperature))
      .keyBy(_._1)
      .timeWindow(Time.seconds(15))
      .process(new HighAndLowTempProcessFunction)

    // 输出结果
    minTempStream.print
  }

  // IN, OUT, KEY, W <: Window
  class HighAndLowTempProcessFunction extends ProcessWindowFunction[(String, Double), MinMaxTemp, String, TimeWindow] {
    override def process(key: String, context: Context, elements: Iterable[(String, Double)], out: Collector[MinMaxTemp]) = {
      // 过滤温度
      val tmps = elements.map(_._2)
      // 获取窗口
      val end = context.window.getEnd
      out.collect(MinMaxTemp(key, tmps.max, tmps.min, end))
    }
  }

}
